package bullgo

import (
	"bullgo/config"
	"bullgo/readcommand"
	"context"
	"fmt"
	"math"
	"reflect"
	"strings"
	"time"

	"github.com/go-redis/redis/v8"
	uuid "github.com/satori/go.uuid"
)

type timerManger struct {
	rootCtx  *context.Context
	rootDone *context.CancelFunc
	timers   map[string]timeStruct
}

type timeStruct struct {
	timer *time.Timer
	done  *context.CancelFunc
}

func (t *timerManger) Set(name string, timeDuration time.Duration, fn func()) {
	if t.timers == nil {
		t.timers = make(map[string]timeStruct)
	}
	ts, ok := t.timers[name]
	if ok {
		(*ts.done)()
		ts.timer.Stop()
	}
	timer := time.NewTimer(timeDuration)
	ctx, Done := context.WithCancel(context.Background())
	t.timers[name] = timeStruct{
		timer, &Done,
	}
	go func(rootCtx context.Context, ctx context.Context) {
		select {
		case <-timer.C:
			fn()
		case <-ctx.Done():
			return
		case <-rootCtx.Done():
			return
		}
	}(*t.rootCtx, ctx)
}

func (t *timerManger) Clear(name string) {
	tt, ok := t.timers[name]
	if ok {
		(*tt.done)()
		tt.timer.Stop()
		delete(t.timers, name)
	}
}

func (t *timerManger) ClearAll() {
	(*t.rootDone)()
	t.timers = make(map[string]timeStruct)
}

type tickerManger struct {
	rootCtx  *context.Context
	rootDone *context.CancelFunc
	tickers  map[string]tickerStruct
}

type tickerStruct struct {
	ticker *time.Ticker
	done   *context.CancelFunc
}

func (t *tickerManger) Set(name string, timeDuration time.Duration, fn func()) {
	if t.tickers == nil {
		t.tickers = make(map[string]tickerStruct)
	}
	tt, ok := t.tickers[name]
	if ok {
		(*tt.done)()
		tt.ticker.Stop()
	}
	ticker := time.NewTicker(timeDuration)
	ctx, Done := context.WithCancel(context.Background())
	t.tickers[name] = tickerStruct{
		ticker, &Done,
	}
	go func(rootCtx context.Context, ctx context.Context) {
		for {
			select {
			case <-ticker.C:
				fn()
			case <-ctx.Done():
				return
			case <-rootCtx.Done():
				return
			}
		}
	}(*t.rootCtx, ctx)
}

func (t *tickerManger) Clear(name string) {
	tt, ok := t.tickers[name]
	if ok {
		(*tt.done)()
		tt.ticker.Stop()
		delete(t.tickers, name)
	}
}

func (t *tickerManger) ClearAll() {
	(*t.rootDone)()
	t.tickers = make(map[string]tickerStruct)
}

// MinimumRedisVersion ...
const MinimumRedisVersion = "2.8.18"

//
const (
	Active       = "active"
	Delayed      = "delayed"
	Drained      = "drained"
	Empty        = ""
	ID           = "id"
	MetaPaused   = "metaPaused"
	Limiter      = "limiter"
	Paused       = "paused"
	Priority     = "priority"
	Stalled      = "stalled"
	Wait         = "wait"
	Completed    = "completed"
	Failed       = "failed"
	StalledCheck = "stalled-check"
	Repeat       = "repeat"
)

// QueueKeys ...
type QueueKeys struct {
	Active       string
	Delayed      string
	Drained      string
	Empty        string
	ID           string
	MetaPaused   string
	Limiter      string
	Paused       string
	Priority     string
	Stalled      string
	Wait         string
	Completed    string
	Failed       string
	StalledCheck string
	Repeat       string
}

// JobHandler ...
type JobHandler func(job *Job, done func(err error, returnValue interface{}))

// MaxTimeoutMS ... 32 bit signed
var MaxTimeoutMS = math.Pow(2, 31) - 1

// Queue ...
type Queue struct {
	queueName        string
	token            string
	keyPrefix        string
	queueKeys        QueueKeys
	queueSettings    config.QueueSettings
	queueLimiter     *QueueLimiter
	bclient          *redis.Client
	eclient          *redis.Client
	redisOptions     redis.Options
	Paused           bool
	TimerManger      timerManger
	TickerManger     tickerManger
	delayedTimestamp int64
	Drained          bool
	Event
	*readcommand.ExtendRedis
}

// QueueLimiter ...
type QueueLimiter struct {
	max      int64
	duration int64
}

// NewQueue ... init queue
func NewQueue(queueName string, cnf config.Config) *Queue {
	q := new(Queue)
	ctxT, DoneT := context.WithCancel(context.Background())
	q.TimerManger.rootCtx = &ctxT
	q.TimerManger.rootDone = &DoneT
	ctxTT, DoneTT := context.WithCancel(context.Background())
	q.TickerManger.rootCtx = &ctxTT
	q.TickerManger.rootDone = &DoneTT
	q.delayedTimestamp = math.MaxInt64
	q.queueName = queueName
	q.token = uuid.NewV4().String()
	if cnf.KeyPrefix == "" {
		q.keyPrefix = "bullgo"
	} else {
		q.keyPrefix = cnf.KeyPrefix
	}
	q.queueSettings = config.QueueSettings{
		LockDuration:      30000,
		MaxStalledCount:   1,
		StalledInterval:   30000,
		LockRenewTime:     10000,
		RetryProcessDelay: 5000,
		DrainDelay:        5,
		GuardInterval:     5000,
	}
	if cnf.QueueSettings != nil {
		v1 := reflect.ValueOf(q.queueSettings)
		v1elem := reflect.ValueOf(&q.queueSettings)
		for i := 0; i < v1.NumField(); i++ {
			field2 := reflect.ValueOf(*cnf.QueueSettings).Field(i)
			if !reflect.DeepEqual(field2.Interface(), reflect.Zero(field2.Type()).Interface()) {
				if !v1elem.Elem().Field(i).CanSet() {
				} else {
					v1elem.Field(i).Set((field2))
				}
			}
		}
	}

	c := redis.NewClient(cnf.RedisOptions)
	ExtendRedis, err := readcommand.NewExtendRedis(context.Background(), c)
	if err != nil {
		fmt.Printf("error: %v", err)
	}
	q.ExtendRedis = ExtendRedis
	q.bclient = redis.NewClient(cnf.RedisOptions)
	q.eclient = redis.NewClient(cnf.RedisOptions)
	q.redisOptions = *cnf.RedisOptions

	// 转换queuekey
	baseQueueKey := QueueKeys{
		Active, Delayed, Drained,
		Empty, ID, MetaPaused,
		Limiter, Paused, Priority,
		Stalled, Wait, Completed,
		Failed, StalledCheck, Repeat,
	}
	v1 := reflect.ValueOf(baseQueueKey)
	v1elem := reflect.ValueOf(&baseQueueKey).Elem()
	for i := 0; i < v1.NumField(); i++ {
		v1elem.Field(i).Set(reflect.ValueOf(q.Tokey(v1.Field(i))))
	}
	q.queueKeys = baseQueueKey
	return q
}

// Tokey ... 转换为redis存储key
func (queue *Queue) Tokey(key interface{}) string {
	return strings.Join([]string{queue.keyPrefix, queue.queueName, fmt.Sprint(key)}, ":")
}

// Add ...
func (queue *Queue) Add(data map[string]interface{}) (*Job, error) {
	return queue.AddWithOption(data, AddJobOption{})
}

// AddWithOption ...
func (queue *Queue) AddWithOption(data map[string]interface{}, opts AddJobOption) (*Job, error) {
	job := new(Job)
	jobID, err := job.createWithOption(queue, data, opts)
	if err != nil {
		return nil, err
	}
	job.ID = *jobID
	job.LockKey = job.Queue.Tokey(*jobID) + ":lock"
	return job, nil
}

// Empty ...
func (queue *Queue) Empty() error {
	pipe := queue.Client.TxPipeline()
	qkeys := queue.queueKeys
	pipe.LRange(queue.Ctx, qkeys.Wait, 0, -1)
	pipe.LRange(queue.Ctx, qkeys.Paused, 0, -1)
	pipe.ZRange(queue.Ctx, qkeys.Delayed, 0, -1)
	pipe.Del(queue.Ctx, qkeys.Wait)
	pipe.Del(queue.Ctx, qkeys.Paused)
	pipe.Del(queue.Ctx, qkeys.MetaPaused)
	pipe.Del(queue.Ctx, qkeys.Delayed)
	pipe.Del(queue.Ctx, qkeys.Priority)
	if _, err := queue.Client.Ping(queue.Ctx).Result(); err != nil {
		panic(err)
	}
	cmders, err := pipe.Exec(queue.Ctx)
	if err != nil {
		return err
	}

	var ids []string
	waitIDs, _ := cmders[0].(*redis.StringSliceCmd).Result()
	pausedIDs, _ := cmders[1].(*redis.StringSliceCmd).Result()
	delayIDs, _ := cmders[2].(*redis.StringSliceCmd).Result()
	ids = append(pausedIDs, waitIDs...)
	ids = append(ids, delayIDs...)
	for i := 0; i < len(ids); i++ {
		pipe.Del(queue.Ctx, queue.Tokey(ids[i]))
	}
	_, err = pipe.Exec(queue.Ctx)
	if err != nil {
		return err
	}
	return nil
}

func (queue *Queue) GetBClient() *redis.Client {
	if queue.bclient != nil {
		return queue.bclient
	}
	queue.bclient = redis.NewClient(&queue.redisOptions)
	return queue.bclient
}

func (queue *Queue) GetEClient() *redis.Client {
	if queue.eclient != nil {
		return queue.eclient
	}
	queue.eclient = redis.NewClient(&queue.redisOptions)
	return queue.eclient
}

// Close ...
func (queue *Queue) Close() error {
	var err error
	err = queue.ExtendRedis.Close()
	if queue.bclient != nil && err == nil {
		err = queue.bclient.Close()
	}
	if queue.eclient != nil && err == nil {
		err = queue.eclient.Close()
	}

	if err != nil {
		return err
	}
	return nil
}

// Pause ...
func (queue *Queue) Pause(isLocal bool) (bool, error) {
	if isLocal {
		queue.Paused = true
	}

	_, err := pause(queue, true).Result()
	if err != nil {
		return false, err
	}
	return true, nil
}

// Resume ...
func (queue *Queue) Resume(isLocal bool) (bool, error) {
	if isLocal {
		queue.Paused = false
	}

	_, err := pause(queue, false).Result()
	if err != nil {
		return false, err
	}
	return true, nil
}

// Clean ...
func (queue *Queue) Clean(timeAgo int64, set string) (bool, error) {
	_, err := cleanJobsInSet(queue, set, time.Now().UnixNano()/1e6-timeAgo, 0).Result()
	if err != nil {
		return false, err
	}
	return true, nil
}

// CleanWithLimit ...
func (queue *Queue) CleanWithLimit(timeAgo int64, set string, limit int) (bool, error) {
	_, err := cleanJobsInSet(queue, set, time.Now().UnixNano()/1e6-timeAgo, limit).Result()
	if err != nil {
		return false, err
	}
	return true, nil
}
